package org.example.realtime.traffic.dwd.log.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.example.realtime.jtp.common.utils.JdbcUtil;
import org.example.realtime.jtp.common.utils.KafkaUtil;
import org.example.realtime.traffic.dwd.log.entity.TrafficFlowType;
import org.example.realtime.traffic.dwd.log.function.FlowTypeReportFunction;

import java.time.Duration;

/**
 * @Title: JobTrafficTotalFlowTypeJob
 * @Author Lianzy
 * @Package org.example.realtime.traffic.dwd.log.job
 * @Date 2025/5/29 18:03
 * @description  卡口流量排行  车两类型分布
 */
public class JobTrafficTotalFlowTypeJob {
    public static void main(String[] args) throws Exception {

        // 1 - 创建执行环境 - env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 2 - 数据源 - source
        DataStream<String> trafficStream = KafkaUtil.consumerKafka(env, "traffic_events");
//        trafficStream.print("flow");

        // 3 - 数据转换 - transformation
        DataStream<String> resultStream = handle(trafficStream);
//        resultStream.print();

        // 4 - 数据接收器 - sink
        JdbcUtil.sinkToClickhouseUpsert(resultStream,
                "INSERT INTO traffic_monitoring.dws_traffic_flow_type_report_total(\n" +
                        "    window_start_time, window_end_time,\n" +
                        "    camera_id, vehicle_type,\n" +
                        "    vehicle_type_count,\n" +
                        "    ts\n" +
                        ")\n" +
                        "VALUES (?, ?, ?, ?, ?, ?)"
        );

        // 5 - 执行任务 - execute
        env.execute("JobTrafficTotalFlowJob");

    }

    private static DataStream<String> handle(DataStream<String> pageStream) {
        // todo - 将stream中的每条日志数据封装实体类对象Bean

        SingleOutputStreamOperator<TrafficFlowType> cameraIdBeanStream = pageStream.map(new RichMapFunction<String, TrafficFlowType>() {

            @Override
            public TrafficFlowType map(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);

                String cameraId = jsonObject.getString("cameraId");
                String vehicleType = jsonObject.getString("vehicleType");
                Long vehicleTypeCount = 1L;
//                String  licensePlateCount = jsonObject.getString("license_plate_count");

                return new TrafficFlowType(
                        null,
                        null,
                        cameraId,
                        vehicleType,
                        vehicleTypeCount,
                        jsonObject.getLong("ts")
                );
            }
        });
//        cameraIdBeanStream.print();

        // todo  - 设置 水位线  事件时间字段
        SingleOutputStreamOperator<TrafficFlowType> timeStream = cameraIdBeanStream.assignTimestampsAndWatermarks(WatermarkStrategy
                .<TrafficFlowType>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                .withTimestampAssigner(
                        new SerializableTimestampAssigner<TrafficFlowType>() {
                            @Override
                            public long extractTimestamp(TrafficFlowType element, long recordTimestamp) {
                                return element.getTs();
                            }
                        }
                )
        );

//        timeStream.print();

        // todo  - 分组
        KeyedStream<TrafficFlowType, String> trafficFlowStringKeyedStream = timeStream.keyBy(value -> {
            return value.getCameraId() + "," + value.getVehicleType();
        });

        // todo  - 开窗 : 滚动窗口,滚动窗口大小为 1 分钟
        WindowedStream<TrafficFlowType, String, TimeWindow> windowStream = trafficFlowStringKeyedStream.window(
                TumblingEventTimeWindows.of(Time.minutes(1))
        );


        // todo  - 聚合 : 对窗口中的数据进行聚合


        SingleOutputStreamOperator<String> reportStream = windowStream.apply(new FlowTypeReportFunction());






        return reportStream;
    }
}
